package com.dianwoba.universe.redis.operation.impl;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.dianwoba.universe.cache.JedisCallback;
import com.dianwoba.universe.cache.JedisTemplate;
import com.dianwoba.universe.redis.operation.HashOperations;

import redis.clients.jedis.BinaryJedisCommands;

/**
 * Default implementation of {@link HashOperations}.
 * 
 * @author Costin Leau
 * @author Christoph Strobl
 */
public class DefaultHashOperations<K, HK, HV, X extends BinaryJedisCommands> 
    extends AbstractOperations<K, Object, X>  implements HashOperations<K, HK, HV> {

	@SuppressWarnings("unchecked")
	public DefaultHashOperations(JedisTemplate<K, ?, X> template) {
		super((JedisTemplate<K, Object, X>) template);
	}

	@SuppressWarnings("unchecked")
	public HV get(K key, Object hashKey) {
		final byte[] rawKey = rawKey(key);
		final byte[] rawHashKey = rawHashKey(hashKey);

		byte[] rawHashValue = execute(new JedisCallback<byte[], X>() {

			public byte[] doInRedis(X jedis) {
				return jedis.hget(rawKey, rawHashKey);
			}
		});

		return (HV) deserializeHashValue(rawHashValue);
	}

	public Boolean hasKey(K key, Object hashKey) {
		final byte[] rawKey = rawKey(key);
		final byte[] rawHashKey = rawHashKey(hashKey);

		return execute(new JedisCallback<Boolean, X>() {

			public Boolean doInRedis(X jedis) {
				return jedis.hexists(rawKey, rawHashKey);
			}
		});
	}

	public Long increment(K key, HK hashKey, final long delta) {
		final byte[] rawKey = rawKey(key);
		final byte[] rawHashKey = rawHashKey(hashKey);

		return execute(new JedisCallback<Long, X>() {

			public Long doInRedis(X jedis) {
				return jedis.hincrBy(rawKey, rawHashKey, delta);
			}
		});

	}

	public Double increment(K key, HK hashKey, final double delta) {
		final byte[] rawKey = rawKey(key);
		final byte[] rawHashKey = rawHashKey(hashKey);

		return execute(new JedisCallback<Double, X>() {
			public Double doInRedis(X jedis) {
				return jedis.hincrByFloat(rawKey, rawHashKey, delta);
			}
		});
	}

	public Set<HK> keys(K key) {
		final byte[] rawKey = rawKey(key);

		Set<byte[]> rawValues = execute(new JedisCallback<Set<byte[]>, X>() {

			public Set<byte[]> doInRedis(X jedis) {
				return jedis.hkeys(rawKey);
			}
		});

		return deserializeHashKeys(rawValues);
	}

	public Long size(K key) {
		final byte[] rawKey = rawKey(key);

		return execute(new JedisCallback<Long, X>() {

			public Long doInRedis(X jedis) {
				return jedis.hlen(rawKey);
			}
		});
	}

	public void putAll(K key, Map<? extends HK, ? extends HV> m) {
		if (m.isEmpty()) {
			return;
		}

		final byte[] rawKey = rawKey(key);

		final Map<byte[], byte[]> hashes = new LinkedHashMap<byte[], byte[]>(m.size());

		for (Map.Entry<? extends HK, ? extends HV> entry : m.entrySet()) {
			hashes.put(rawHashKey(entry.getKey()), rawHashValue(entry.getValue()));
		}

		execute(new JedisCallback<Object, X>() {

			public Object doInRedis(X jedis) {
				jedis.hmset(rawKey, hashes);
				return null;
			}
		});
	}

	public List<HV> multiGet(K key, Collection<HK> fields) {
		if (fields.isEmpty()) {
			return Collections.emptyList();
		}

		final byte[] rawKey = rawKey(key);

		final byte[][] rawHashKeys = new byte[fields.size()][];

		int counter = 0;
		for (HK hashKey : fields) {
			rawHashKeys[counter++] = rawHashKey(hashKey);
		}

		List<byte[]> rawValues = execute(new JedisCallback<List<byte[]>, X>() {

			public List<byte[]> doInRedis(X jedis) {
				return jedis.hmget(rawKey, rawHashKeys);
			}
		});

		return deserializeHashValues(rawValues);
	}

	public void put(K key, HK hashKey, HV value) {
		final byte[] rawKey = rawKey(key);
		final byte[] rawHashKey = rawHashKey(hashKey);
		final byte[] rawHashValue = rawHashValue(value);

		execute(new JedisCallback<Object, X>() {

			public Object doInRedis(X jedis) {
				jedis.hset(rawKey, rawHashKey, rawHashValue);
				return null;
			}
		});
	}

	public Long putIfAbsent(K key, HK hashKey, HV value) {
		final byte[] rawKey = rawKey(key);
		final byte[] rawHashKey = rawHashKey(hashKey);
		final byte[] rawHashValue = rawHashValue(value);

		return execute(new JedisCallback<Long, X>() {

			public Long doInRedis(X jedis) {
				return jedis.hsetnx(rawKey, rawHashKey, rawHashValue);
			}
		});
	}

	public List<HV> values(K key) {
		final byte[] rawKey = rawKey(key);

		List<byte[]> rawValues = execute(new JedisCallback<List<byte[]>, X>() {

			public List<byte[]> doInRedis(X jedis) {
				return new ArrayList<byte[]>(jedis.hvals(rawKey));
			}
		});

		return deserializeHashValues(rawValues);
	}

	public void delete(K key, Object... hashKeys) {
		final byte[] rawKey = rawKey(key);
		final byte[][] rawHashKeys = rawHashKeys(hashKeys);

		execute(new JedisCallback<Object, X>() {

			public Object doInRedis(X jedis) {
				jedis.hdel(rawKey, rawHashKeys);
				return null;
			}
		});
	}

	public Map<HK, HV> entries(K key) {
		final byte[] rawKey = rawKey(key);

		Map<byte[], byte[]> entries = execute(new JedisCallback<Map<byte[], byte[]>, X>() {

			public Map<byte[], byte[]> doInRedis(X jedis) {
				return jedis.hgetAll(rawKey);
			}
		});

		return deserializeHashMap(entries);
	}
}
